package com.hdl.client.consumer;

import com.hdl.JSONSerializer;
import com.hdl.RpcEncoder;
import com.hdl.RpcRequest;
import com.hdl.client.handler.NettyClientHandler;
import com.hdl.client.handler.NettyClientHandlerSingleton;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.timeout.ReadTimeoutHandler;
import io.netty.handler.timeout.WriteTimeoutHandler;
import org.apache.curator.framework.CuratorFramework;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.data.Stat;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.net.URLDecoder;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;

/**
 * 类名 ServiceDiscover
 * 描述 服务发现
 *
 * @author hedonglin
 * @version 1.0
 * @date 2020/6/17 17:03
 */

@Component
public class ServiceDiscover {
    /**
     * 服务提供者节点
     */
    public static String SERVICE_PROVIDER = "/service_provider";
    public static List<NettyClientHandler> clientHandlers = new ArrayList<>();

    @Autowired(required = false)
    private CuratorFramework zkClient;

    /**
     * 服务数据
     */
    private volatile List<String> dataList = new ArrayList<>();
    /**
     * 服务发现
     * @return 服务器地址
     */
    public String discover() {
        String data = null;
        int size = dataList.size();
        if (size > 0) {
            if (size == 1) {
                data = dataList.get(0);
            } else {
                //随机获取其中的一个
                data = dataList.get(ThreadLocalRandom.current().nextInt(size));
            }
        }
        return data;
    }

    public void watchNode(String path){
        List<String> list = new ArrayList<>();
        try {
            Stat stat = zkClient.checkExists().forPath(SERVICE_PROVIDER);
            // 如果不存在，则创建路径
            if (stat == null){
                zkClient.create().forPath(SERVICE_PROVIDER);
            }
            System.out.println("节点监听中....");
            list = zkClient.getChildren()
                    .usingWatcher((Watcher) watchedEvent -> {
                        System.out.println("数据变更通知");
                        if (watchedEvent.getType() == Watcher.Event.EventType.NodeChildrenChanged){
                            dataList.clear();
                            // 继续监听自己，并处理业务逻辑
                            watchNode(SERVICE_PROVIDER);
                        }
                    }).forPath(path);
            System.out.println("服务器节点个数变化：" + list.size());
            for (String s : list) {
                String decodeStr = URLDecoder.decode(s, StandardCharsets.UTF_8);
                dataList.add(decodeStr);
                System.out.println("服务器：" + decodeStr);
            }
//            String server = discover();
            if (list.size() == 0){
                System.out.println("没有获取到RPC远程服务器");
                return;
            }
            // 初始化客户端
            try {
                // 这里本应该是有下线的只剔除下线的，有上线的只初始化上线的。这里简单处理，直接全部重新初始化
                initClient(list);
            } catch (Exception e) {
                e.printStackTrace();
                System.out.println("初始化Netty客户端失败");
            }
        } catch (Exception e) {
            e.printStackTrace();
            System.out.println("zookeeper节点监听异常");
        }

    }
    public void initClient(List<String> connectStrings) {
        clientHandlers.clear();
        for (String connectString : connectStrings) {
            // 为每个服务器连接创建一个Handler
            NettyClientHandler client = new NettyClientHandler();
            // 解码
            String decodeStr = URLDecoder.decode(connectString, StandardCharsets.UTF_8);
            // 设置绑定的服务器信息
            client.setConnectString(decodeStr);
            clientHandlers.add(client);
            // 初始化
            initClient(decodeStr, client);
        }

    }



    /*** 初始化客户端 */
    public void initClient(String connectString, NettyClientHandler client) {
        // 分割
        String[] split = connectString.split(":");
        EventLoopGroup group = new NioEventLoopGroup();
        Bootstrap b = new Bootstrap();
        b.group(group)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                // 设置连接超时时间
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
                // 返回数据超时时间
                .option(ChannelOption.SO_TIMEOUT, 5000)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    public void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline p = ch.pipeline();
//                        //读取超时 在设置时间内没有读取操作（5s内没有读取到服务端数据则超时）
//                        p.addLast(new ReadTimeoutHandler(5000, TimeUnit.MILLISECONDS));
//                        p.addLast(new WriteTimeoutHandler(5000, TimeUnit.MILLISECONDS));
                        p.addLast(new RpcEncoder(RpcRequest.class, new JSONSerializer()));
                        p.addLast(new StringDecoder());
                        p.addLast(client);
                    }
                });
        try {
            b.connect(split[0], Integer.parseInt(split[1])).sync();
        } catch (InterruptedException e) {
            System.out.println("服务器连接失败");
            e.printStackTrace();
            // 异常时关闭连接池
            group.shutdownGracefully();
        }
    }


}
